package regular2MR;

import java.io.IOException;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BooleanWritable;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

public class QueryEval {

	public static void main(String[] args) throws IOException,
			ClassNotFoundException, InterruptedException {
		try {

			boolean isRunOnServer = false;

			long startTime = System.currentTimeMillis();

			Configuration conf = new Configuration();

			String input = "/user/quyet/input";
			String output = "/user/quyet/output";

			if (isRunOnServer) {
				conf.addResource(new Path(
						"/usr/local/hadoop-1.1.1-bin/conf/core-site.xml"));
				conf.addResource(new Path(
						"/usr/local/hadoop-1.1.1-bin/conf/hdfs-site.xml"));
				input = args[0];
				output = args[1];
				Ultility.LOCAL_GRAPH_QUERY = args[2];
				Ultility.HDFS_GRAPH_QUERY = args[3];
			} else {
				conf.addResource(new Path(
						"/usr/local/hadoop/conf/core-site.xml"));
				conf.addResource(new Path(
						"/usr/local/hadoop/conf/hdfs-site.xml"));
				Ultility.LOCAL_GRAPH_QUERY = "/home/quyet/test/query/automaton-4-4.txt";
				Ultility.HDFS_GRAPH_QUERY = "/user/quyet/automaton/graphquery.txt";
			}

			Ultility.cacheGraphQuery(conf);

			Job job = new Job(conf, "QueryEval");
			job.setJarByClass(QueryEval.class);

			job.setMapperClass(EvalMapper.class);
			job.setReducerClass(EvalReducer.class);

			job.setMapOutputKeyClass(IntWritable.class);
			job.setMapOutputValueClass(RvecWritable.class);

			job.setOutputKeyClass(IntWritable.class);
			job.setOutputValueClass(BooleanWritable.class);

			job.setInputFormatClass(GraphFileInputFormat.class);
			GraphFileInputFormat.addInputPath(job, new Path(input));

			// FileInputFormat.addInputPath(job, new Path(input));
			FileOutputFormat.setOutputPath(job, new Path(output));

			job.waitForCompletion(true);

			long stopTime = System.currentTimeMillis();
			long elapsedTime = stopTime - startTime;

			System.out.println("Finished after: " + elapsedTime / 1000 + " s\n");

		} catch (Exception e) {
			e.printStackTrace();
		}

		// System.exit(job.waitForCompletion(true) ? 0 : 1);
	}

}
